package cc.eddic.practice.concurrency.impl;

import cc.eddic.practice.concurrency.LogItem;
import cc.eddic.practice.concurrency.LogParser;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.stereotype.Component;


import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
@Component
@Qualifier("LogParserWithFutureCxmImpl")
@Slf4j
public class LogParserWithFutureCxmImpl implements LogParser {
    private static final int NUM_THREADS = Runtime.getRuntime().availableProcessors();
    @Override
    public Map<String, List<LogItem>> parse(String folderName) {
        Map<String, List<LogItem>> res=new ConcurrentHashMap<>();
        List<Future<Model>> futures;
        ExecutorService executorService= Executors.newFixedThreadPool(NUM_THREADS);
        try {
            val p= Paths.get(folderName);
            try(val walk= Files.walk(p)){
                futures = walk.parallel()
                        .filter(Files::isRegularFile)
                        .filter(path -> path.getFileName().toString().endsWith(".log"))
                        .map(model ->executorService.submit(()->parseFile(model)))
                        .collect(Collectors.toList());}
            }
            catch (IOException e) {
                log.error("File does not exist : {}",e.getMessage());
                return res;
            }

        if (log.isInfoEnabled()) log.info("{} futures is running", futures.size());
        allOf(futures);

        if (log.isInfoEnabled()) log.info("get result");
        futures.stream().parallel().map(this::getFuture).filter(Objects::nonNull)
                .forEach(x -> res.put(x.getPath() .toString(),x.getItems()));

        return res;
    }

    private static <T> void allOf(final List<Future<T>> futures){
        while(true){
            val b=futures.stream().anyMatch(x->!x.isDone());
            if(!b) break;
            try {
                Thread.sleep(10);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
            if(log.isDebugEnabled()) log.debug("{} futures done",futures.stream().filter(Future::isDone));
        }
    }



    private Model getFuture(Future<Model> future){
        try {
            return future.get();
        } catch (InterruptedException | ExecutionException e) {
            log.error(e.getMessage(),e);
            return null;
        }
    }

    protected Model parseFile(Path path){
        try (val lines = Files.lines(path)) {
            val items = lines.map(new LogItemParserCxmImpl()::parse)
                    .filter(Optional::isPresent)
                    .map(Optional::get)
                    .collect(Collectors.toList());
            if(log.isDebugEnabled()) log.debug("PARSE : {} => {}", path, items.size());
            return new Model(path, items);
        } catch (IOException ex) {
            log.error(ex.getMessage(), ex);
        }
        return new Model(path);
    }

    @Getter
    protected static class Model{

        public Model(Path path){this(path,new ArrayList<>());}

        public Model(Path path,List<LogItem> logItems){
            this.path=path;
            this.items=logItems;
        }
        private final Path path;
        private final List<LogItem> items;
    }

}

